package com.pine.dbant.common.plugin;


import com.pine.dbant.common.exeception.BusinessException;
import com.pine.dbant.common.exeception.ProviderException;
import com.pine.dbant.common.model.ConnectionInfo;
import com.pine.dbant.common.model.SQLStatement;
import com.pine.dbant.common.model.SqlCommandType;
import com.pine.dbant.common.sql.MultiSqlParser;
import com.pine.sunflower.core.plugin.Configuration;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author xiaoyuan
 * @create 2022/11/4 17:38
 **/
@Slf4j
public abstract class AbstractJdbcConnector extends AbstractConnector {

    protected Connection connection;

    private  final Set<String> queryCommandTypes = new HashSet<>();

    protected AbstractJdbcConnector() {


    }

    public AbstractJdbcConnector(Configuration configuration, Connection connection, ConnectionInfo connectionInfo) {
        super(configuration, connectionInfo);
        this.connection = connection;
        try {
            this.connection.setAutoCommit(false);
        } catch (SQLException e) {
            throw new JdbcException("取消自动提交错误");
        }
        queryCommandTypes.addAll(Arrays.asList(SqlCommandType.SELECT.name(),
                SqlCommandType.EXPLAIN.name(),SqlCommandType.SHOW.name(),SqlCommandType.WITH.name()));
        configQueryCommandType(queryCommandTypes);
    }

    @Override
    public void execute(ExecuteParams params, ExecuteCallback callback) throws ProviderException {
        List<SQLStatement> sqlStatements;
        try {
            sqlStatements = parseStatements(params.getScript());
        } catch (Exception e) {
            throw new BusinessException("sql脚本语法错误", e);
        }
        try{
            callback.beforeExecute(sqlStatements);
            ExecuteParams executeParams;
            ExecuteRecord executeRecord;
            for (SQLStatement statement : sqlStatements) {
                if(StringUtils.isBlank(statement.getSql())){
                    continue;
                }
                log.debug("执行sql：{}",statement.getSql());
                callback.beforeOneExecute(statement);
                executeParams = params.copy();
                executeParams.setScript(statement.getSql());
                Long startTime = System.currentTimeMillis();
                if (queryCommandTypes.contains(statement.getType())) {
                    if (!(executeParams instanceof QueryExecuteParams)) {
                        executeParams = new QueryExecuteParams();
                        executeParams.setScript(statement.getSql());
                    }
                    executeRecord = doQuery(statement, (QueryExecuteParams) executeParams);
                } else {
                    executeRecord = doExecute(statement, executeParams);
                }
                if(executeRecord==null){
                    continue;
                }
                executeRecord.setTime(System.currentTimeMillis()-startTime);
                callback.processRow(executeRecord);
                callback.afterOneExecute(statement, executeRecord);
            }
        }finally {
            commit();
            close();
        }

    }

    private String  getCommandType(String sql){
        for(int i=0;i<sql.length();i++){
            if(sql.charAt(i) == ' '){
                return sql.substring(0,i);
            }
        }
        return sql;
    }

    public abstract ExecuteRecord doExecute(SQLStatement statement, ExecuteParams params) throws ProviderException;

    public abstract QueryExecuteRecord doQuery(SQLStatement statement, QueryExecuteParams params) throws ProviderException;


    public List<SQLStatement> parseStatements(String sql) {
        try {
            List<String> sqls  =  new MultiSqlParser().runStatements(sql);
            return sqls.stream().map(s -> new SQLStatement(s,getCommandType(s.trim())
                    .toUpperCase())).collect(Collectors.toList());
        } catch (Exception e) {
            throw new BusinessException("sql解析失败");
        }
    }


    @Override
    public void commit() throws ProviderException {
        if (connection == null) {
            throw new IllegalStateException("Attempt to commit a transaction on a closed connection");
        }

        flush();

        try {
            connection.commit();
        } catch (Exception e) {
            e.printStackTrace();
           // throw new JdbcException("Unable to commit transaction", e);
        }
    }

    public void flush() throws ProviderException {

    }

    @Override
    public void rollback() throws ProviderException, UnsupportedOperationException {
        if (connection == null) {
            throw new IllegalStateException("Attempt to roll back a transaction on a closed connection");
        }
        try {
            connection.rollback();
        } catch (Exception e) {
            throw new JdbcException("Unable to roll back transaction", e);
        }
    }

    @Override
    public void close() throws ProviderException {
        if (connection != null) {
            try {
                connection.close();
                connection = null;
            } catch (SQLException e) {
                throw new JdbcException("Unable to close a connection", e);
            }

        }
    }

    public void  configQueryCommandType(Set<String> queryCommandTypes ){

    }




    @Override
    public Connection getNativeConnection() {
        return this.connection;
    }
}
